/**  
  -- The Performance Driven Lime Project -- 

  A self adaptive tuple space for mobile agents and hosts by dynamic
  adapting tuple's replication policy.
  
  Copyright (C) 2008
  Xing Jiankuan, Pan Wenhui
  SAGroup of Tsinghua University

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */
package sagroup.dracots;

import java.util.ArrayList;
import java.util.logging.Logger;

import lime2.Location;
import lime2.TupleID;
import sagroup.dracots.policy.PolicyRegistrar;
import sagroup.dracots.policy.RPTuple;
import sagroup.dracots.policy.ReplicationPolicy;


/**
 * Replication Policy Switcher. Responsible of executing replication policy switch.
 * The switch process will start a new temporarily thread to finish the detailed 
 * switch jobs.
 * @author Xing Jiankuan
 *
 */
public class PolicySwitcher {
	
	private DracoTS dts;
	
	private PolicyRegistrar policyRegistrar;
	
	private ReplicaRequestRegistrar replicaRegistrar;
	
	private Logger LOG = Logger.getLogger(this.getClass().getName());
	
	public PolicySwitcher(DracoTS dts) {
		this.dts = dts;
		policyRegistrar = dts.getPolicyRegistrar();
		replicaRegistrar = dts.getReplicaRequestRegistrar();
	}

	public void switchPolicy(TupleType type, Location node, ReplicationPolicy newPolicy) {
	
		ReplicationPolicy oldPolicy = policyRegistrar.getReplicaPolicy(type,
				node);

		if (oldPolicy == null)
			oldPolicy = ReplicationPolicy.SO; // null is equivalent to SO

		if (!oldPolicy.equals(newPolicy)) { // if different, perform policy
											// switch
			// first, accept it
			LOG.info(String.format("SWITCH,%s,%s,%s,%s", node, type, oldPolicy, newPolicy));

			policyRegistrar.updateReplicaPolicy(type, node, newPolicy);

			// then, switch old policy to new policy switch
			if (newPolicy.equals(ReplicationPolicy.FR)) {
				// RC->FR or SO->FR
				new SwitchToFRThread(replicaRegistrar, type, node).start();
			} else if (newPolicy.equals(ReplicationPolicy.SO)) {
				// FR->SO or RC->SO
				new SwitchToSOThread(type, node).start();
			} else if (newPolicy.equals(ReplicationPolicy.RC)) {
				if (oldPolicy.equals(ReplicationPolicy.FR)) {
					// FR->RC
					new SwitchFRToRCThread(replicaRegistrar, type, node)
							.start();
				}
				// switch from SO to RC do nothing
			}
		}
		// If the initial policy is SO, we also need to add it to RP hash map
		else if (oldPolicy.equals(ReplicationPolicy.SO)) {
			policyRegistrar.updateReplicaPolicy(type, node, oldPolicy);
		}
	}

	class SwitchToSOThread extends Thread {

		TupleType type;

		Location node;

		public SwitchToSOThread(TupleType type, Location node) {
			this.type = type;
			this.node = node;
		}

		public void run() {
			switchToSO(type, node);
		}

		private void switchToSO(TupleType type, Location node) {
			// for active side, eliminate all the replicas requests and
			// replicas themselves with that type
			if (node.equals(Location.HERE)) {
				// create a template that can match any replica with that type
				ShareTuple template = new ShareTuple(null, Location.HERE,
						Location.HERE, type, null, null, Location.UNSPECIFIED,
						Location.UNSPECIFIED, true);
				template.setMatchAll(true);
				DracoTuple[] replicas = dts.lrr.rrdg(template);
				if (replicas == null)
					return; // not found, do nothing
				TupleID[] replicaIDs = new TupleID[replicas.length];
				for (int i = 0; i < replicas.length; i++) {
					replicaIDs[i] = replicas[i].getVID();
				}
				dts.removeReplica(replicaIDs);
			}
			// the passive side will degregister replica requests
			// when they receive the "RemoveReplicaRequest" command
		}
	}

	class SwitchToFRThread extends Thread {

		ReplicaRequestRegistrar replicaRegistrar;

		TupleType type;

		Location node;

		public SwitchToFRThread(ReplicaRequestRegistrar replicaRegistrar,
				TupleType type, Location node) {
			this.replicaRegistrar = replicaRegistrar;
			this.type = type;
			this.node = node;
		}

		public void run() {
			switchToFR(replicaRegistrar, type, node);
		}

		private void switchToFR(ReplicaRequestRegistrar replicaRegistrar,
				TupleType type, Location node) {
			if (!node.equals(Location.HERE)) {
				// the passive side should do:
				// Replicate and push all local tuples, which are of
				// specified type and haven't build a replication on the
				// active side, to the active side.

				// TODO: Here only consider the MASTER policy, so
				// template only specifies cur, dest, type and
				// is_replica.
				
				// define a template that can match all local masters with that
				// type
				ShareTuple template = new ShareTuple(null, Location.HERE,
						Location.HERE, type, null, null, Location.UNSPECIFIED,
						Location.UNSPECIFIED, false);
				template.setMatchAll(true);
				DracoTuple[] masters = dts.rdg(template);
				if (masters != null) {
					ArrayList replicas = new ArrayList();
					for (int i = 0; i < masters.length; ++i) {
						if (replicaRegistrar.getReplicaVersion(masters[i]
								.getVID(), node) == null && !(masters[i] instanceof RPTuple)) {
							// if the replica request is not found, add it
							// Also, if the master is the RPTuple, push is not needed since it 
							// has got the RP tuple from remote 
							ShareTuple replica = ((ShareTuple) masters[i])
									.replicate(node);
							replicas.add(replica);
						}
					}
					if (replicas.size() == 0)
						return; // nothing to push
					ShareTuple[] st_tmp = new ShareTuple[replicas.size()];
					st_tmp = (ShareTuple[]) replicas.toArray(st_tmp);
					dts.pushReplica(st_tmp, node);
				}
			} // do nothing for the active side
		}
	}

	class SwitchFRToRCThread extends Thread {

		ReplicaRequestRegistrar replicaRegistrar;

		TupleType type;

		Location node;

		public SwitchFRToRCThread(ReplicaRequestRegistrar replicaRegistrar,
				TupleType type, Location node) {
			this.replicaRegistrar = replicaRegistrar;
			this.type = type;
			this.node = node;
		}

		public void run() {
			switchFRtoRC(replicaRegistrar, type, node);
		}

		private void switchFRtoRC(ReplicaRequestRegistrar replicaRegistrar,
				TupleType type, Location node) {
			// for active side, remove all the local replicas with that type
			if (node.equals(Location.HERE)) {
				TupleID[] toBeRemoved = dts.getUnusedReplicaByType(type);
				if (toBeRemoved.length == 0)
					return;
				else
					dts.removeReplica(toBeRemoved);
			}
		}
	}
}
